"""
文件名: redis.py
描述: Redis缓存连接和操作管理模块

本模块包含以下主要功能：
1. Redis异步连接管理
2. 缓存操作封装（get、set、delete等）
3. 分布式锁支持
4. 会话存储管理
5. 数据序列化和反序列化

依赖模块:
   - redis: Redis客户端库
   - aioredis: 异步Redis客户端
   - json: JSON序列化
   - pickle: Python对象序列化
   - typing: 类型注解

使用示例:
   >>> from src.core.redis import get_redis, RedisManager
   >>> redis_manager = get_redis()
   >>> await redis_manager.set("key", "value", expire=3600)
   >>> value = await redis_manager.get("key")
   >>> 
   >>> # 使用分布式锁
   >>> async with redis_manager.lock("resource_key"):
   ...     # 执行需要锁定的操作
   ...     pass

注意事项:
   - Redis连接需要在应用启动时初始化
   - 使用连接池提高性能
   - 注意缓存过期时间设置
   - 大量数据建议使用Pipeline批量操作

作者: AI助手
创建日期: 2024-06-27
最后修改: 2024-06-27
版本: 1.0.0
"""

import json
import pickle
from contextlib import asynccontextmanager
from typing import Any, Dict, List, Optional, Union, AsyncGenerator

import redis.asyncio as redis
from redis.asyncio import Redis, ConnectionPool
from redis.exceptions import RedisError, ConnectionError, TimeoutError

from .config import get_settings
from .logger import get_logger

logger = get_logger(__name__)
settings = get_settings()

# 全局Redis连接实例
redis_client: Optional[Redis] = None


class RedisManager:
    """Redis管理器类"""
    
    def __init__(self):
        self.client: Optional[Redis] = None
        self.connection_pool: Optional[ConnectionPool] = None
        self._initialized = False
    
    async def initialize(self) -> None:
        """
        初始化Redis连接
        
        配置包括：
        1. 创建连接池
        2. 配置连接参数
        3. 设置编码和解码
        4. 测试连接可用性
        """
        if self._initialized:
            logger.warning("Redis已经初始化，跳过重复初始化")
            return
        
        try:
            # 创建连接池
            self.connection_pool = ConnectionPool.from_url(
                settings.redis_url,
                max_connections=20,  # 最大连接数
                retry_on_timeout=True,  # 超时重试
                socket_connect_timeout=5,  # 连接超时（秒）
                socket_timeout=5,  # 数据传输超时（秒）
                health_check_interval=30,  # 健康检查间隔（秒）
                encoding="utf-8",
                decode_responses=False,  # 手动处理编码
            )
            
            # 创建Redis客户端
            self.client = Redis(
                connection_pool=self.connection_pool,
                socket_connect_timeout=5,
                socket_timeout=5,
                retry_on_timeout=True,
                encoding="utf-8",
                decode_responses=False,
            )
            
            # 设置全局变量
            global redis_client
            redis_client = self.client
            
            # 测试连接
            await self.client.ping()
            
            self._initialized = True
            logger.info("Redis初始化成功", redis_url=settings.redis_url.split("@")[-1])
            
        except Exception as e:
            logger.error(f"Redis初始化失败: {e}", error=str(e))
            raise
    
    async def close(self) -> None:
        """关闭Redis连接"""
        if self.client:
            await self.client.close()
        if self.connection_pool:
            await self.connection_pool.disconnect()
        logger.info("Redis连接已关闭")
    
    async def health_check(self) -> bool:
        """
        Redis健康检查
        
        Returns:
            bool: Redis是否健康
        """
        if not self.client:
            return False
        
        try:
            await self.client.ping()
            return True
        except Exception as e:
            logger.error(f"Redis健康检查失败: {e}", error=str(e))
            return False
    
    async def get(self, key: str, default: Any = None) -> Any:
        """
        获取缓存值
        
        Args:
            key: 缓存键
            default: 默认值
            
        Returns:
            Any: 缓存值或默认值
        """
        if not self.client:
            return default
        
        try:
            value = await self.client.get(key)
            if value is None:
                return default
            
            # 尝试JSON解码
            try:
                return json.loads(value.decode("utf-8"))
            except (json.JSONDecodeError, UnicodeDecodeError):
                # 如果JSON解码失败，尝试pickle解码
                try:
                    return pickle.loads(value)
                except (pickle.PickleError, ValueError):
                    # 都失败则返回原始字符串
                    return value.decode("utf-8", errors="ignore")
                    
        except Exception as e:
            logger.error(f"Redis获取失败: {e}", key=key, error=str(e))
            return default
    
    async def set(
        self,
        key: str,
        value: Any,
        expire: Optional[int] = None,
        nx: bool = False,
        xx: bool = False,
    ) -> bool:
        """
        设置缓存值
        
        Args:
            key: 缓存键
            value: 缓存值
            expire: 过期时间（秒）
            nx: 仅当键不存在时设置
            xx: 仅当键存在时设置
            
        Returns:
            bool: 是否设置成功
        """
        if not self.client:
            return False
        
        try:
            # 序列化值
            if isinstance(value, (dict, list, tuple)):
                serialized_value = json.dumps(value, ensure_ascii=False).encode("utf-8")
            elif isinstance(value, str):
                serialized_value = value.encode("utf-8")
            elif isinstance(value, (int, float, bool)):
                serialized_value = json.dumps(value).encode("utf-8")
            else:
                # 复杂对象使用pickle序列化
                serialized_value = pickle.dumps(value)
            
            # 设置缓存
            result = await self.client.set(
                key, serialized_value, ex=expire, nx=nx, xx=xx
            )
            return bool(result)
            
        except Exception as e:
            logger.error(f"Redis设置失败: {e}", key=key, error=str(e))
            return False
    
    async def delete(self, *keys: str) -> int:
        """
        删除缓存键
        
        Args:
            *keys: 要删除的键列表
            
        Returns:
            int: 删除的键数量
        """
        if not self.client or not keys:
            return 0
        
        try:
            return await self.client.delete(*keys)
        except Exception as e:
            logger.error(f"Redis删除失败: {e}", keys=keys, error=str(e))
            return 0
    
    async def exists(self, *keys: str) -> int:
        """
        检查键是否存在
        
        Args:
            *keys: 要检查的键列表
            
        Returns:
            int: 存在的键数量
        """
        if not self.client or not keys:
            return 0
        
        try:
            return await self.client.exists(*keys)
        except Exception as e:
            logger.error(f"Redis检查存在失败: {e}", keys=keys, error=str(e))
            return 0
    
    async def expire(self, key: str, seconds: int) -> bool:
        """
        设置键过期时间
        
        Args:
            key: 缓存键
            seconds: 过期时间（秒）
            
        Returns:
            bool: 是否设置成功
        """
        if not self.client:
            return False
        
        try:
            return await self.client.expire(key, seconds)
        except Exception as e:
            logger.error(f"Redis设置过期时间失败: {e}", key=key, error=str(e))
            return False
    
    async def ttl(self, key: str) -> int:
        """
        获取键剩余生存时间
        
        Args:
            key: 缓存键
            
        Returns:
            int: 剩余秒数，-1表示永不过期，-2表示键不存在
        """
        if not self.client:
            return -2
        
        try:
            return await self.client.ttl(key)
        except Exception as e:
            logger.error(f"Redis获取TTL失败: {e}", key=key, error=str(e))
            return -2
    
    async def incr(self, key: str, amount: int = 1) -> int:
        """
        递增键值
        
        Args:
            key: 缓存键
            amount: 递增量
            
        Returns:
            int: 递增后的值
        """
        if not self.client:
            return 0
        
        try:
            return await self.client.incr(key, amount)
        except Exception as e:
            logger.error(f"Redis递增失败: {e}", key=key, error=str(e))
            return 0
    
    async def decr(self, key: str, amount: int = 1) -> int:
        """
        递减键值
        
        Args:
            key: 缓存键
            amount: 递减量
            
        Returns:
            int: 递减后的值
        """
        if not self.client:
            return 0
        
        try:
            return await self.client.decr(key, amount)
        except Exception as e:
            logger.error(f"Redis递减失败: {e}", key=key, error=str(e))
            return 0
    
    @asynccontextmanager
    async def lock(
        self,
        key: str,
        timeout: int = 10,
        blocking_timeout: int = 10,
    ) -> AsyncGenerator[bool, None]:
        """
        分布式锁上下文管理器
        
        Args:
            key: 锁键名
            timeout: 锁超时时间（秒）
            blocking_timeout: 等待锁的超时时间（秒）
            
        Yields:
            bool: 是否成功获取锁
        """
        if not self.client:
            yield False
            return
        
        lock = self.client.lock(
            name=f"lock:{key}",
            timeout=timeout,
            blocking_timeout=blocking_timeout,
        )
        
        try:
            acquired = await lock.acquire()
            if acquired:
                logger.debug(f"获取分布式锁成功: {key}")
            else:
                logger.warning(f"获取分布式锁失败: {key}")
            yield acquired
        except Exception as e:
            logger.error(f"分布式锁操作失败: {e}", key=key, error=str(e))
            yield False
        finally:
            try:
                if acquired:
                    await lock.release()
                    logger.debug(f"释放分布式锁: {key}")
            except Exception as e:
                logger.error(f"释放分布式锁失败: {e}", key=key, error=str(e))
    
    async def get_info(self) -> Dict[str, Any]:
        """
        获取Redis服务器信息
        
        Returns:
            Dict[str, Any]: Redis服务器信息
        """
        if not self.client:
            return {}
        
        try:
            info = await self.client.info()
            return {
                "version": info.get("redis_version"),
                "connected_clients": info.get("connected_clients"),
                "used_memory": info.get("used_memory_human"),
                "keyspace": {k: v for k, v in info.items() if k.startswith("db")},
            }
        except Exception as e:
            logger.error(f"获取Redis信息失败: {e}", error=str(e))
            return {}


# 全局Redis管理器实例
redis_manager = RedisManager()


async def get_redis_manager() -> RedisManager:
    """
    获取Redis管理器实例
    
    Returns:
        RedisManager: Redis管理器实例
    """
    return redis_manager


async def get_redis() -> Optional[Redis]:
    """
    获取Redis客户端
    
    Returns:
        Optional[Redis]: Redis客户端实例
    """
    return redis_client


async def init_redis() -> None:
    """初始化Redis连接"""
    await redis_manager.initialize()


async def close_redis() -> None:
    """关闭Redis连接"""
    await redis_manager.close()


async def redis_health_check() -> bool:
    """Redis健康检查"""
    return await redis_manager.health_check()


# 便捷函数
async def cache_get(key: str, default: Any = None) -> Any:
    """获取缓存值的便捷函数"""
    return await redis_manager.get(key, default)


async def cache_set(
    key: str,
    value: Any,
    expire: Optional[int] = None,
) -> bool:
    """设置缓存值的便捷函数"""
    return await redis_manager.set(key, value, expire)


async def cache_delete(*keys: str) -> int:
    """删除缓存键的便捷函数"""
    return await redis_manager.delete(*keys)


# 导出主要接口
__all__ = [
    "RedisManager",
    "redis_manager",
    "get_redis_manager",
    "get_redis",
    "init_redis",
    "close_redis",
    "redis_health_check",
    "cache_get",
    "cache_set",
    "cache_delete",
]